#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <fcntl.h>
#include <errno.h>
#include <ibase.h>
#include <time.h>
#include <sys/stat.h>
#include <sys/mman.h>
#include <sys/time.h>
#include "mmtrie.h"
#include "logger.h"
#include "qsmap.h"
#include "iqueue.h"
#include "db.h"
#include "mmdb.h"
#define  M_QMAP_NAME    "mmdb.qmap"
#define  M_LOG_NAME     "mmdb.log"
#define  M_STATE_NAME   "mmdb.state"
#define  M_QTASK_NAME   "mmdb.qtask"
#define  M_QPAGE_NAME   "mmdb.qpage"
#define  M_DB_DIR       "cache"
#ifndef  LL
#define  LL(x) ((long long int)x)
#endif
#ifdef MAP_LOCKED
#define MMAP_SHARED MAP_SHARED|MAP_LOCKED
#else
#define MMAP_SHARED MAP_SHARED
#endif
/* over merge */
int mmdb_over_merge(MMDB *mmdb, int qid, int pid, IQSET *qset, IRECORD *recs, int *nrecs);
int pmkdir(char *path)
{
    struct stat st;
    char fullpath[M_PATH_MAX];
    char *p = NULL;
    int level = -1, ret = -1;

    if(path)
    {
        strcpy(fullpath, path);
        p = fullpath;
        while(*p != '\0')
        {
            if(*p == '/' )
            {
                while(*p != '\0' && *p == '/' && *(p+1) == '/')++p;
                if(level > 0)
                {
                    *p = '\0';                   
                    memset(&st, 0, sizeof(struct stat));
                    ret = stat(fullpath, &st);
                    if(ret == 0 && !S_ISDIR(st.st_mode)) return -1;
                    if(ret != 0 && mkdir(fullpath, 0755) != 0) return -1;
                    *p = '/';
                }
                level++;
            }
            ++p;
        }
        return 0;
    }
    return -1;
}

/* set base dir */
int mmdb_set_basedir(MMDB *mmdb, char *basedir)
{
    char path[M_PATH_MAX];
    struct stat st = {0};
    SNODE *nodes = NULL;
    int i = 0;

    /* qmap */
    sprintf(path, "%s/%s", basedir, M_QMAP_NAME);
    pmkdir(path);
    if((mmdb->qmap = mmtrie_init(path)) == NULL)
    {
        fprintf(stderr, "Initialize mmtrie(%s) failed, %s\n", path, strerror(errno));
        _exit(-1);
        return -1;
    }
    /* state */
    sprintf(path, "%s/%s", basedir, M_STATE_NAME);
    if((mmdb->stateio.fd = open(path, O_CREAT|O_RDWR, 0644)) > 0
            && fstat(mmdb->stateio.fd, &st) == 0)
    {
        if(st.st_size < sizeof(MSTATE))
        {
                if(ftruncate(mmdb->stateio.fd, sizeof(MSTATE))) _exit(-1);
        }
        if((mmdb->stateio.map = mmap(NULL, sizeof(MSTATE), PROT_READ|PROT_WRITE,
            MMAP_SHARED, mmdb->stateio.fd, 0)) && mmdb->stateio.map != (void *)-1)
        {
            mmdb->state = (MSTATE *)mmdb->stateio.map;
            mmdb->stateio.size = sizeof(MSTATE);
            if((nodes = mmdb->state->nodes))
            {
                for(i = 0; i < M_NODES_MAX; i++)
                {
                    if(nodes[i].status > 0)
                    {
                        iqueue_push(mmdb->queue, i);
                        //np = &i;QUEUE_PUSH(mmdb->queue, int, np);
                    }
                }
            }
        }
        else
        {
            fprintf(stderr, "mmap state file (%s) failed, %s\n", path, strerror(errno));
            _exit(-1);
        }
    }
    else
    {
        fprintf(stderr, "open node file(%s) failed, %s\n", path, strerror(errno));
        _exit(-1);
    }
    /* qtask */
    sprintf(path, "%s/%s", basedir, M_QTASK_NAME);
    if((mmdb->qtaskio.fd = open(path, O_CREAT|O_RDWR, 0644)) > 0
            && fstat(mmdb->qtaskio.fd, &st) == 0)
    {
        mmdb->qtaskio.end = st.st_size;
        mmdb->qtaskio.size = M_QTASK_MAX * sizeof(QTASK);
        if(st.st_size == 0)
        {
            mmdb->qtaskio.end = mmdb->qtaskio.size;
            if(ftruncate(mmdb->qtaskio.fd,  mmdb->qtaskio.end) != 0)
            {
                fprintf(stderr, "ftruncate %s failed, %s\n", path, strerror(errno));
                _exit(-1);
            }
        }
        else if(st.st_size > mmdb->qtaskio.size)
        {
            mmdb->qtaskio.end = mmdb->qtaskio.size = st.st_size;
        }
        if((mmdb->qtaskio.map = mmap(NULL, mmdb->qtaskio.size, PROT_READ|PROT_WRITE,
            MMAP_SHARED, mmdb->qtaskio.fd, 0)) == NULL || mmdb->qtaskio.map == (void *)-1)
        {
            fprintf(stderr, "mmap record file (%s) failed, %s\n", path, strerror(errno));
            _exit(-1);
        }
    }
    else
    {
        fprintf(stderr, "open record file(%s) failed, %s\n", path, strerror(errno));
        _exit(-1);
    }
    /* qpage */
    sprintf(path, "%s/%s", basedir, M_QPAGE_NAME);
    if((mmdb->qpageio.fd = open(path, O_CREAT|O_RDWR, 0644)) > 0
            && fstat(mmdb->qpageio.fd, &st) == 0)
    {
        mmdb->qpageio.end = st.st_size;
        mmdb->qpageio.size = M_QPAGE_MAX * sizeof(QPAGE);
        if(st.st_size == 0)
        {
            mmdb->qpageio.end = mmdb->qpageio.size;
            if(ftruncate(mmdb->qpageio.fd,  mmdb->qpageio.end) != 0)
            {
                fprintf(stderr, "ftruncate %s failed, %s\n", path, strerror(errno));
                _exit(-1);
            }
        }
        else if(st.st_size > mmdb->qpageio.size)
        {
            mmdb->qpageio.end = mmdb->qpageio.size = st.st_size;
        }
        if((mmdb->qpageio.map = mmap(NULL, mmdb->qpageio.size, PROT_READ|PROT_WRITE,
            MMAP_SHARED, mmdb->qpageio.fd, 0)) == NULL || mmdb->qpageio.map == (void *)-1)
        {
            fprintf(stderr, "mmap record file (%s) failed, %s\n", path, strerror(errno));
            _exit(-1);
        }
    }
    else
    {
        fprintf(stderr, "open record file(%s) failed, %s\n", path, strerror(errno));
        _exit(-1);
    }
    /* db */
    sprintf(path, "%s/%s", basedir, M_DB_DIR);
    mmdb->db = db_init(path, 0);
    /* logger */
    sprintf(path, "%s/%s", basedir, M_LOG_NAME);
    LOGGER_INIT(mmdb->logger, path);
    //gettimeofday(&tv, NULL);
    mmdb->start_time = time(NULL);//(off_t)tv.tv_sec * (off_t)1000000 + (off_t)tv.tv_usec;
    return 0;
}

/* mmdb set cache life time (seconds) */
int mmdb_set_cache_life_time(MMDB *mmdb, int cache_life_time)
{
    if(mmdb)
    {
        MUTEX_LOCK(mmdb->mutex);
        if(mmdb->state)
        {
            mmdb->state->cache_life_time = cache_life_time;
        }
        MUTEX_UNLOCK(mmdb->mutex);
        return 0;
    }
    return -1;
}

/* add node */
int mmdb_add_node(MMDB *mmdb, int type, char *ip, int port, int limit)
{
    int i = 0, n = 0, nodeid = 0;
    char buf[M_LINE_SIZE];
    SNODE *nodes = NULL;

    if(mmdb && mmdb->state && type > 0 && ip && port > 0 
            && (n = sprintf(buf, "%d:%s:%d", type, ip, port)) > 0
            && (nodeid = (mmtrie_get(mmdb->qmap, buf, n) - 1)) < 0
            && (nodes = mmdb->state->nodes))
    {
        MUTEX_LOCK(mmdb->mutex);
        for(i = 0; i < M_NODES_MAX; i++)
        {
            if(nodes[i].status == 0)
            {
                strcpy(nodes[i].ip, ip);
                nodes[i].port = port;
                nodes[i].type = type;
                nodes[i].limit = limit;
                nodes[i].status = 1;
                mmtrie_add(mmdb->qmap, buf, n, i + 1);
                nodeid = i;
                mmdb->state->nnodes++;
                iqueue_push(mmdb->queue, i);
                //np = &i;QUEUE_PUSH(mmdb->queue, int, np);
                break;
            }
        }
        MUTEX_UNLOCK(mmdb->mutex);
    }
    return nodeid;
}

/* del node */
int mmdb_del_node(MMDB *mmdb, int nodeid)
{
    char buf[M_LINE_SIZE];
    int ret = -1, n = 0;
    SNODE *nodes = NULL;

    if(mmdb && mmdb->state && nodeid >= 0 && nodeid < M_NODES_MAX)
    {
        MUTEX_LOCK(mmdb->mutex);
        if((nodes = mmdb->state->nodes) && nodes[nodeid].status > 0 
                && (n = sprintf(buf, "%d:%s:%d", nodes[nodeid].type, 
                        nodes[nodeid].ip, nodes[nodeid].port)) > 0)
        {
            mmtrie_del(mmdb->qmap, buf, n);
            nodes[nodeid].status = 0;
            --(mmdb->state->nnodes);
            ret = nodeid;
        }
        MUTEX_UNLOCK(mmdb->mutex);
    }
    return ret;
}

/* update node */
int mmdb_update_node(MMDB *mmdb, int nodeid, int limit)
{
    SNODE *nodes = NULL;
    int ret = -1;

    if(mmdb && mmdb->state && nodeid >= 0 && nodeid < M_NODES_MAX)
    {
        MUTEX_LOCK(mmdb->mutex);
        if((nodes = mmdb->state->nodes) && nodes[nodeid].status > 0) 
        {
            nodes[nodeid].limit = limit;
            ret = nodeid;
        }
        MUTEX_UNLOCK(mmdb->mutex);
    }
    return ret;
}

/* pop node */
int mmdb_pop_node(MMDB *mmdb, SNODE *node)
{
    SNODE *nodes = NULL;
    int nodeid = -1;

    if(mmdb && mmdb->state && node && QTOTAL(mmdb->queue) > 0)
    {
        MUTEX_LOCK(mmdb->mutex);
        //np = &nodeid;
        //if(QUEUE_POP(mmdb->queue, int, np) == 0 
        iqueue_pop(mmdb->queue, &nodeid);
        if(nodeid >= 0 && nodeid < M_NODES_MAX
                && (nodes = mmdb->state->nodes))
        {
            memcpy(node, &(nodes[nodeid]), sizeof(SNODE));
        }
        else nodeid = -1;
        MUTEX_UNLOCK(mmdb->mutex);
    }
    return nodeid;
}

/* list nodes */
int mmdb_list_nodes(MMDB *mmdb, char *s, char *es)
{
    SNODE *nodes  = NULL;
    int i = 0, n = 0;
    char *p = NULL;

    if(mmdb && s && mmdb->state && (nodes = mmdb->state->nodes))
    {
        MUTEX_LOCK(mmdb->mutex);
        if(mmdb->state->nnodes > 0 && s < es)
        {
            p = s;
            p += sprintf(p, "({'count':'%d', 'nodes':{", mmdb->state->nnodes);
            for(i = 0; i < M_NODES_MAX; i++)
            {
                if(nodes[i].status > 0 && s < es)
                {
                    p += sprintf(p, "'%d':{'type':'%d', 'limit':'%d','ip':'%s', 'port':'%d'},",
                            i, nodes[i].type, nodes[i].limit, nodes[i].ip, nodes[i].port);
                }
            }
            --p;
            p += sprintf(p, "}})");
            n = p - s;
        }
        MUTEX_UNLOCK(mmdb->mutex);
    }
    return n;
}

/* mmdb pop smap */
void *mmdb_pop_smap(MMDB *mmdb)
{
    void *smap = NULL;
    int x = 0;

    if(mmdb)
    {
        if(mmdb->nqsmaps > 0)
        {
            x = --(mmdb->nqsmaps);
            smap = mmdb->qsmaps[x];
            mmdb->qsmaps[x] = NULL;
        }
        else
        {
            smap = QSMAP_INIT();
        }
    }
    return smap;
}

/* push smap */
void mmdb_push_smap(MMDB *mmdb, void *smap)
{
    int x = 0;

    if(mmdb && smap)
    {
        if(mmdb->nqsmaps < M_MAPS_MAX)
        {
            QSMAP_RESET(smap);
            x = mmdb->nqsmaps++;
            mmdb->qsmaps[x] = smap;
        }
        else
        {
            QSMAP_CLEAN(smap);
        }
    }
    return ;
}

/* mmdb pop qres */
void *mmdb_pop_qres(MMDB *mmdb)
{
    void *qres = NULL;
    int x = 0;

    if(mmdb)
    {
        if(mmdb->nqqres > 0)
        {
            x = --(mmdb->nqqres);
            qres = mmdb->qqres[x];
            mmdb->qqres[x] = NULL;
        }
        else
        {
            qres = calloc(1, sizeof(QRES));
        }
    }
    return qres;
}

/* push qres */
void mmdb_push_qres(MMDB *mmdb, void *qres)
{
    int x = 0;

    if(mmdb && qres)
    {
        if(mmdb->nqqres < M_QRES_MAX)
        {
            memset(qres, 0, sizeof(QRES));
            x = mmdb->nqqres++;
            mmdb->qqres[x] = qres;
        }
        else
        {
            free(qres);
        }
    }
    return ;
}

#define CHECK_QTASKIO(mmdb, rid)                                                            \
do                                                                                          \
{                                                                                           \
    if((off_t)(rid * sizeof(QTASK)) >= mmdb->qtaskio.end)                                   \
    {                                                                                       \
        mmdb->qtaskio.end = (off_t)((rid / M_QTASK_INCREMENT)+1);                           \
        mmdb->qtaskio.end *= (off_t)(sizeof(QTASK) * M_QTASK_INCREMENT);                    \
        if(ftruncate(mmdb->qtaskio.fd, mmdb->qtaskio.end) != 0)break;                       \
    }                                                                                       \
    if(mmdb->qtaskio.end > mmdb->qtaskio.size)                                              \
    {                                                                                       \
        if(mmdb->qtaskio.map && mmdb->qtaskio.size > 0)                                     \
        {                                                                                   \
            munmap(mmdb->qtaskio.map, mmdb->qtaskio.size);                                  \
        }                                                                                   \
        mmdb->qtaskio.map = NULL;                                                           \
        mmdb->qtaskio.size = mmdb->qtaskio.end;                                             \
    }                                                                                       \
    if(mmdb->qtaskio.map == NULL)                                                           \
    {                                                                                       \
        if((mmdb->qtaskio.map = mmap(NULL, mmdb->qtaskio.size, PROT_READ|PROT_WRITE,        \
                MMAP_SHARED, mmdb->qtaskio.fd, 0)) == NULL                                  \
                || mmdb->qtaskio.map == (void *)-1)                                         \
        break;                                                                              \
    }                                                                                       \
}while(0)

/* query taskid */
int mmdb_qid(MMDB *mmdb, char *s, char *es)
{
    int qid = -1;

    if(mmdb && mmdb->qmap)
    {
        if((qid = mmtrie_xadd(MMTR(mmdb->qmap), s, (es - s))) > 0)
        {
            MUTEX_LOCK(mmdb->mutex);
            if(qid > mmdb->state->qid_max) 
            {
                CHECK_QTASKIO(mmdb, qid);
                mmdb->state->qid_max = qid;
            }
            MUTEX_UNLOCK(mmdb->mutex);
        }
    }
    return qid;
}

/* query taskid */
int mmdb_qstatus(MMDB *mmdb, int qid, int *new_query)
{
    QTASK *qtasks = NULL;
    int status = -1;

    if(mmdb && qid <= mmdb->state->qid_max)
    {
        MUTEX_LOCK(mmdb->mutex);
        if((qtasks = (QTASK *)mmdb->qtaskio.map))
        {
            if(qtasks[qid].mod_time < mmdb->start_time)
            {
                qtasks[qid].mod_time = time(NULL);
                qtasks[qid].status = M_STATUS_QUERY;
                qtasks[qid].left = 0;
                qtasks[qid].qres = NULL;
                qtasks[qid].map = NULL;
                status = M_STATUS_FREE;
                *new_query = 1;
            }
            else
            {
                status = qtasks[qid].status;
                if((qtasks[qid].mod_time + mmdb->state->cache_life_time) < time(NULL)
                        && qtasks[qid].status == M_STATUS_FREE)
                {
                    *new_query = 1;
                    qtasks[qid].status = M_STATUS_QUERY;
                }
            }
        }
        MUTEX_UNLOCK(mmdb->mutex);
    }
    return status;
}

/* reset status */
int mmdb_reset_qstatus(MMDB *mmdb, int qid)
{
    QTASK *qtasks = NULL;
    int ret = -1;

    if(mmdb && qid > 0 && qid <= mmdb->state->qid_max)
    {
        MUTEX_LOCK(mmdb->mutex);
        if((qtasks = (QTASK *)mmdb->qtaskio.map))
        {
            qtasks[qid].status = M_STATUS_FREE;
            qtasks[qid].mod_time = 0;
            ret = 0;
        }
        MUTEX_UNLOCK(mmdb->mutex);
    }
    return ret;
}

/* mmdb set qparser */
int mmdb_set_query(MMDB *mmdb, int qid, IQUERY *query, int nquerys)
{
    QTASK *qtasks = NULL;
    QRES *qres = NULL;
    int ret = -1;

    if(mmdb && qid > 0 && qid <= mmdb->state->qid_max && query)
    {
        MUTEX_LOCK(mmdb->mutex);
        if(qid != query->qid)
        {
            FATAL_LOGGER(mmdb->logger, "qid:%d query->qid:%d query->ntop:%d nterms:%d", qid, query->qid, query->ntop, query->nqterms);
            goto end;

        }
        if((qtasks = (QTASK *)(mmdb->qtaskio.map)))
        {
            if((qres = qtasks[qid].qres) == NULL)
            {
                qtasks[qid].qres = qres = mmdb_pop_qres(mmdb);
                if(qres)memset(&(qres->qset), 0, sizeof(IQSET));
            }
            qtasks[qid].left = nquerys;
            qtasks[qid].ntop = query->ntop;
            if(nquerys == 0)
            {
                qtasks[qid].status = M_STATUS_FREE;
                qtasks[qid].mod_time = 0;
            }
            if(qres)
            {
                qres->qset.nqterms = query->nqterms;
                memcpy(&(qres->qset.qterms), &(query->qterms), 
                        sizeof(QTERM) * query->nqterms);
                memcpy(qres->qset.displaylist, query->display, 
                        sizeof(IDISPLAY) * IB_FIELDS_MAX);
                ret = qid;
            }
        }
end:
        MUTEX_UNLOCK(mmdb->mutex);
    }
    return ret;
}

/* mmdb over merge */
int mmdb_over_merge(MMDB *mmdb, int qid, int pid, IQSET *qset, IRECORD *recs, int *nrecs)
{
    int i = 0, x = 0, n = 0, to = 0; 
    void *dp = NULL, *map = NULL;
    unsigned int score = 0;
    char buf[M_LINE_SIZE];
    QTASK *qtasks = NULL;
    QPAGE *qpages = NULL;
    QRES *qres = NULL;
    long id = 0;

    if(mmdb && qid > 0 && qid <= mmdb->state->qid_max && (qtasks = (QTASK *)(mmdb->qtaskio.map)) 
            && (map = qtasks[qid].map) && (qres = qtasks[qid].qres))
    {
        //save records
        i = 0;
        while(PQS(map)->count > 0)
        {
            QSMAP_POP_RMAX(map, score, dp);
            //fprintf(stdout, "%s::%d %u:%u\n", __FILE__, __LINE__, (unsigned int )dp, score);
            if((id = ((long )dp)) > 0 && i < IB_TOPK_NUM)
            {
                qres->records[i].score = score;
                qres->records[i].globalid = id;
            }
            ++i;
        }
        if(i > 0)
        {
            if(i > IB_TOPK_NUM) i = IB_TOPK_NUM;
            qres->qset.res.qid = qid;
            qres->qset.res.count = i;
            n = sprintf(buf, "q:%d", qid);
            qtasks[qid].recid = db_xset_data(PDB(mmdb->db), buf, n, 
                    (char *)qtasks[qid].qres, sizeof(IQSET) + sizeof(IRECORD) * i);
            if(pid > 0 && pid <= mmdb->state->pid_max && qset && recs && nrecs
                    && (qpages = (QPAGE *)(mmdb->qpageio.map)))
            {
                memcpy(qset, &(qres->qset), sizeof(IQSET));
                x = qpages[pid].from;
                to = x + qpages[pid].count;
                if(to > qset->res.count) n = qset->res.count - x;
                else n = qpages[pid].count;
                if((qset->count = n) > 0)
                {
                    memcpy(recs, &(qres->records[x]), sizeof(IRECORD) * n);
                    *nrecs = n;
                }
            }
            //gettimeofday(&tv, NULL);
        }
        //(off_t)tv.tv_sec * (off_t)100000 + (off_t)tv.tv_usec;
        mmdb_push_qres(mmdb, qtasks[qid].qres);
        mmdb_push_smap(mmdb, qtasks[qid].map);
        qtasks[qid].qres = NULL;
        qtasks[qid].map = NULL;
        qtasks[qid].mod_time = time(NULL);
        qtasks[qid].status = M_STATUS_FREE;
        return i;
    }
    return 0;
}

/* over query /merge result */
int mmdb_merge(MMDB *mmdb, int qid, int nodeid, IRES *res, IRECORD *records, 
        int pid, IQSET *qset, IRECORD *recs, int *nrecs)
{
    void *map = NULL, *dp = NULL, *olddp = NULL;
    unsigned int score = 0;
    int i = 0, ret = -1;
    QTASK *qtasks = NULL;
    QRES *qres = NULL;

    if(mmdb && qid > 0 && qid <= mmdb->state->qid_max && res && qset && recs)
    {
        if(qid != res->qid)
        {
            FATAL_LOGGER(mmdb->logger, "Invalid qid:%d node:%d to res->qid:%d res:%p records:%p", qid, nodeid, res->qid, res, records);
            return -1;
        }
        MUTEX_LOCK(mmdb->mutex);
        if((qtasks = (QTASK *)mmdb->qtaskio.map))
        {
            if((map = qtasks[qid].map) == NULL)
                qtasks[qid].map = map = mmdb_pop_smap(mmdb);
            if((qres = qtasks[qid].qres) == NULL)
            {
                qtasks[qid].qres = qres = mmdb_pop_qres(mmdb);
            }
            if(res->count > 0 && records && map && qres)
            {
                if(qtasks[qid].ntop == 0)
                {
                    qtasks[qid].ntop = IB_TOPK_NUM;
                    FATAL_LOGGER(mmdb->logger, "map[%p]->count:%d ntop:%d left:%d status:%d records:%p node:%d res[%p]->qid:%d res->count:%d ", map, PQS(map)->count, qtasks[qid].ntop, qtasks[qid].left, qtasks[qid].status, records, nodeid,res,  res->qid, res->count);
                }
                for(i = 0; i < res->count; i++)
                {
                    if(PQS(map)->count > 0 && PQS(map)->count >= qtasks[qid].ntop 
                            && records[i].score < PQS_MINK(map)) break;
                    dp = (void *)((long)(records[i].globalid));
                    if(PQS(map)->count == 0 || PQS(map)->count < qtasks[qid].ntop)
                    {
                        QSMAP_ADD(map, records[i].score, dp, olddp);
                    }
                    else
                    {
                        if(PQS(map)->count > 0 && records[i].score > PQS_MINK(map))
                        {
                            QSMAP_POP_RMIN(map, score, olddp);
                            QSMAP_ADD(map, records[i].score, dp, olddp);
                        }
                    }
                }
                /* merge count */
                qres->qset.res.total += res->total; 
                qres->qset.res.doctotal += res->doctotal; 
                if(res->io_time > qres->qset.res.io_time) 
                    qres->qset.res.io_time = res->io_time;
                if(res->sort_time > qres->qset.res.sort_time) 
                    qres->qset.res.sort_time = res->sort_time;
            }
            /* merge groups */
            if(res->ncatgroups > 0)
            {
                for(i = 0; i < IB_CATGROUPS_MAX; i++)
                {
                    if(res->catgroups[i] > 0)
                    {
                        if(qres->qset.res.catgroups[i] == 0)
                            qres->qset.res.ncatgroups++;
                        qres->qset.res.catgroups[i] += res->catgroups[i];
                    }
                }
            }
            if(--(qtasks[qid].left) == 0) mmdb_over_merge(mmdb, qid, pid, qset, recs, nrecs);
            //DEBUG_LOGGER(mmdb->logger, "qid:%d left:%d", qid, qtasks[qid].left);
            ret = qtasks[qid].left;
        }
        MUTEX_UNLOCK(mmdb->mutex);
    }
    return ret;
}
#define CHECK_QPAGEIO(mmdb, rid)                                                            \
do                                                                                          \
{                                                                                           \
    if((off_t)(rid * sizeof(QPAGE)) >= mmdb->qpageio.end)                                   \
    {                                                                                       \
        mmdb->qpageio.end = (off_t)((rid / M_QTASK_INCREMENT)+1);                           \
        mmdb->qpageio.end *= (off_t)(sizeof(QPAGE) * M_QTASK_INCREMENT);                    \
        if(ftruncate(mmdb->qpageio.fd, mmdb->qpageio.end) != 0)break;                       \
    }                                                                                       \
    if(mmdb->qpageio.end > mmdb->qpageio.size)                                              \
    {                                                                                       \
        if(mmdb->qpageio.map && mmdb->qpageio.size > 0)                                     \
        {                                                                                   \
            munmap(mmdb->qpageio.map, mmdb->qpageio.size);                                  \
        }                                                                                   \
        mmdb->qpageio.map = NULL;                                                           \
        mmdb->qpageio.size = mmdb->qpageio.end;                                             \
    }                                                                                       \
    if(mmdb->qpageio.map == NULL)                                                           \
    {                                                                                       \
        if((mmdb->qpageio.map = mmap(NULL, mmdb->qpageio.size, PROT_READ|PROT_WRITE,        \
                MMAP_SHARED, mmdb->qpageio.fd, 0)) == NULL                                  \
                || mmdb->qpageio.map == (void *)-1)                                         \
        break;                                                                              \
    }                                                                                       \
}while(0)

/* pop summary qres */
int mmdb_get_qres(MMDB *mmdb, int pid, IQSET *qset, IRECORD *records)
{
    int qid = -1, i = 0, to = 0, n = 0, ndata = 0;
    char *data = NULL, *s = NULL;
    QTASK *qtasks = NULL;
    QPAGE *qpages = NULL;

    if(mmdb && pid > 0 && pid <= mmdb->state->pid_max && qset && records)
    {
        MUTEX_LOCK(mmdb->mutex);
        if(pid > mmdb->state->pid_max)
        {
            CHECK_QPAGEIO(mmdb, pid);
            mmdb->state->pid_max = pid;
        }
        //fprintf(stdout, "%s::%d OK\n", __FILE__, __LINE__);
        if((qpages = (QPAGE *)mmdb->qpageio.map) && qpages[pid].count > 0 
                && (qid =  qpages[pid].qid) > 0)
        {

            //&& (qpages[pid].mod_time != qtasks[qid].mod_time) /* check is update */
            /* check/read page N records */
            if((qtasks = (QTASK *)mmdb->qtaskio.map) && qtasks[qid].recid > 0
                && qpages[pid].mod_time <= qtasks[qid].mod_time
                && (ndata = db_get_data(PDB(mmdb->db), qtasks[qid].recid, &data)) > 0)
            {
                /* set modified time */
                qpages[pid].mod_time = qtasks[qid].mod_time;
                memcpy(qset, data, sizeof(IQSET));
                i = qpages[pid].from;
                to = i + qpages[pid].count;
                if(to > qset->res.count) n = qset->res.count - i;
                else n = qpages[pid].count;
                if((qset->count = n) > 0)
                {
                    s = data + sizeof(IQSET) + qpages[pid].from * sizeof(IRECORD);
                    memcpy(records, s, sizeof(IRECORD) * n);
                }
                db_free_data(PDB(mmdb->db), data);
            }
        }
        MUTEX_UNLOCK(mmdb->mutex);
    }
    return n;
}

/* check task */
int mmdb_pid(MMDB *mmdb, int qid, int from, int count)
{
    char line[M_LINE_SIZE];
    QPAGE *qpages = NULL;
    int pid = -1, n = 0;

    if(mmdb && qid > 0 && qid <= mmdb->state->qid_max)
    {
        if((n = sprintf(line, "%d:%d-%d", qid, from, count)) > 0
            && (pid = db_data_id(PDB(mmdb->db), line, n)) > 0)
        {
            MUTEX_LOCK(mmdb->mutex);
            if(pid > mmdb->state->pid_max) 
            {
                CHECK_QPAGEIO(mmdb, pid);
                mmdb->state->pid_max = pid;
            }
            if((qpages = (QPAGE *)mmdb->qpageio.map)) 
            {
                if(qpages[pid].count == 0)
                {
                    qpages[pid].from = from;
                    qpages[pid].count = count;
                    qpages[pid].qid  = qid;
                }
            }
            MUTEX_UNLOCK(mmdb->mutex);
        }
    }
    return pid;
}

int mmdb_check_summary_len(MMDB *mmdb, int pid)
{
    QPAGE *qpages = NULL;
    QTASK *qtasks = NULL;
    int n = 0, qid = 0;

    if(mmdb)
    {
        MUTEX_LOCK(mmdb->mutex);
        if(pid <= mmdb->state->pid_max && (qpages = (QPAGE *)mmdb->qpageio.map)
            && (qid = qpages[pid].qid) <= mmdb->state->qid_max
            && (qtasks = (QTASK *)mmdb->qtaskio.map))
        {
            if(mmdb->state->cache_life_time <= 0 || ((qtasks[qid].mod_time > 0 
                && (time(NULL) - qpages[pid].up_time) < mmdb->state->cache_life_time
                && qpages[pid].mod_time == qtasks[qid].mod_time)) )
            {
                n =  db_get_data_len(PDB(mmdb->db), pid);
            }
        }
        MUTEX_UNLOCK(mmdb->mutex);
    }
    return n;
}

/* read summary */
int mmdb_read_summary(MMDB *mmdb, int pid, char **summary)
{

    if(mmdb && pid > 0 && summary)
    {
        return db_get_data(mmdb->db, pid, summary);
    }
    return -1;
}

/* over summary */
int mmdb_set_summary(MMDB *mmdb, int pid, char *summary, int nsummary)
{
    QPAGE *qpages = NULL;
    int ret = -1;

    if(mmdb && pid > 0 && summary && nsummary > 0)
    {
        MUTEX_LOCK(mmdb->mutex);
        if(pid > mmdb->state->pid_max)
        {
            CHECK_QPAGEIO(mmdb, pid);
            mmdb->state->pid_max = pid;
        }
        if((qpages = (QPAGE *)mmdb->qpageio.map) && qpages[pid].count > 0)
        {
            qpages[pid].up_time = time(NULL);
            ret = db_set_data(mmdb->db, pid, summary, nsummary);
        }
        MUTEX_UNLOCK(mmdb->mutex);
    }
    return ret;
}

/* mmdb free summary */
void mmdb_free_summary(MMDB *mmdb, char *summary)
{
    if(mmdb && summary)
    {
        db_free_data(PDB(mmdb->db), summary);
    }
    return ;
}

/* clear cache */
void mmdb_clear_cache(MMDB *mmdb)
{
    if(mmdb)
    {
        MUTEX_LOCK(mmdb->mutex);
        db_destroy(mmdb->db); 
        MUTEX_UNLOCK(mmdb->mutex);
    }
    return ;
}

/* clean mmdb */
void mmdb_clean(MMDB *mmdb)
{
    int i = 0;

    if(mmdb)
    {
        if(mmdb->queue){iqueue_clean(mmdb->queue);}
        if(mmdb->mutex){MUTEX_DESTROY(mmdb->mutex);}
        if(mmdb->stateio.map) munmap(mmdb->stateio.map, mmdb->stateio.size);
        if(mmdb->stateio.fd > 0) close(mmdb->stateio.fd);
        if(mmdb->qtaskio.map) munmap(mmdb->qtaskio.map, mmdb->qtaskio.size);
        if(mmdb->qtaskio.fd > 0) close(mmdb->qtaskio.fd);
        for(i = 0; i < mmdb->nqsmaps; i++){if(mmdb->qsmaps[i])free(mmdb->qsmaps[i]);}
        for(i = 0; i < mmdb->nqqres; i++){if(mmdb->qqres[i])free(mmdb->qqres[i]);}
        if(mmdb->qmap) mmtrie_clean(MMTR(mmdb->qmap));
        LOGGER_CLEAN(mmdb->logger);
        MUTEX_DESTROY(mmdb->mutex);
    }
    return ;
}

/* initialize mmdb */
MMDB *mmdb_init()
{
    MMDB *mmdb = NULL;
    if((mmdb = (MMDB *)calloc(1, sizeof(MMDB))))
    {
        MUTEX_INIT(mmdb->mutex);
        mmdb->queue         = iqueue_init();
        mmdb->set_basedir   = mmdb_set_basedir;
        mmdb->add_node      = mmdb_add_node;
        mmdb->del_node      = mmdb_del_node;
        mmdb->list_nodes    = mmdb_list_nodes;
        mmdb->pop_node      = mmdb_pop_node;
        mmdb->clean         = mmdb_clean;
    }
    return mmdb;
}

#ifdef _DEBUG_MMDB
int main()
{
    char *dir = "/tmp/mmdb", buf[M_BUF_SIZE], *s = NULL, *es = NULL;
    int n = 0, nodeid = 0, id = 0;
    SNODE node = {0};
    MMDB *mmdb = NULL;

    if((mmdb = mmdb_init()))
    {
        mmdb_set_basedir(mmdb, dir);
        mmdb_add_node(mmdb, M_NODE_DOCD, "10.0.6.81", 4936, 256);
        mmdb_add_node(mmdb, M_NODE_INDEXD, "10.0.6.82", 4936, 256);
        mmdb_add_node(mmdb, M_NODE_INDEXD, "10.0.6.83", 4936, 256);
        mmdb_add_node(mmdb, M_NODE_INDEXD, "10.0.6.84", 4936, 256);
        mmdb_add_node(mmdb, M_NODE_INDEXD, "10.0.6.85", 4936, 256);
        /* list */
        s = buf;
        es = buf + M_BUF_SIZE;
        if((n = mmdb_list_nodes(mmdb, s, es)) > 0)
        {
            buf[n] = '\0';
            fprintf(stdout, "%s\n", buf);
        }
        /* pop node */
        while((nodeid = mmdb_pop_node(mmdb, &node)) >= 0)
        {
            fprintf(stdout, "pop_node(%d):host[%s:%d] type[%d] limit[%d]\n", 
                    nodeid, node.ip, node.port, node.type, node.limit);
            id = nodeid;
        }
        /* del node */
        mmdb_del_node(mmdb, id);
        /* list */
        if((n = mmdb_list_nodes(mmdb, s, es)) > 0)
        {
            buf[n] = '\0';
            fprintf(stdout, "%s\n", buf);
        }
        mmdb_clean(mmdb);
    }
}
#endif
